﻿using System;
using System.Collections.Generic;
using System.Configuration;
using System.Text;
using System.Threading.Tasks;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Infrastructure.MQ
{
    /// <summary>
    /// rabbitMQ消息队列帮助类
    /// </summary>
    public static class RabbitMQHelper
    {
        /// <summary>
        /// rabbitmq的服务地址
        /// </summary>
        private static string ConnectionString = Config.ReadSetting("RabbitMQ:HostName");
        /// <summary>
        /// 用户名
        /// </summary>
        private static string USER_NAME = Config.ReadSetting("RabbitMQ:UserName");
        /// <summary>
        /// 密码
        /// </summary>
        private static string PASS_WORD = Config.ReadSetting("RabbitMQ:PassWord");
        /// <summary>
        /// 虚拟机名称
        /// </summary>
        private static string VIRTUAL_HOST = Config.ReadSetting("RabbitMQ:VirtualHost");



        /// <summary>
        /// 异步发送消息
        /// </summary>
        /// <param name="topicChannel">主题通道</param>
        /// <param name="exchangeName">交换名称</param>
        /// <param name="content">文本消息内容</param>
        /// <param name="ququeTimeSpan">队列过期时间(毫秒)</param>
        /// <param name="messageTimeSpan">消息过期时间(毫秒)</param>
        /// <param name="ip">消息队列服务IP</param>
        /// <param name="port">消息队列服务端口</param>
        /// <param name="userName">用户账户</param>
        /// <param name="passWord">用户密码</param>
        /// <param name="VirtualHost">虚拟机</param>
        public static async Task<bool> SendAsync(string topicChannel, string exchangeName, string content, long? ququeTimeSpan = null, long? messageTimeSpan = null, string ip = null, int port = 5672, string userName = null, string passWord = null, string VirtualHost = null)
        {
            // 检查配置
            if (string.IsNullOrEmpty(userName))
            {
                if (string.IsNullOrEmpty(USER_NAME))
                    throw new Exception("appsettings.json does not define UserName url section:[RabbitMQ:UserName]");
            }
            if (string.IsNullOrEmpty(passWord))
            {
                if (string.IsNullOrEmpty(PASS_WORD))
                    throw new Exception("appsettings.json does not define PassWord url section:[RabbitMQ:PassWord]");
            }
            if (string.IsNullOrEmpty(VirtualHost))
            {
                if (string.IsNullOrEmpty(VIRTUAL_HOST))
                    throw new Exception("appsettings.json does not define VirtualHost url section:[RabbitMQ:VirtualHost]");
            }
            if (string.IsNullOrEmpty(ip))
            {
                if (string.IsNullOrEmpty(ConnectionString))
                    throw new Exception("appsettings.json does not define HostName url section:[RabbitMQ:HostName]");
            }

            // 异步执行发送
            return await Task.Factory.StartNew(() =>
            {
                return Send(topicChannel, exchangeName, content, ququeTimeSpan, messageTimeSpan, ip, port, userName, passWord, VirtualHost);
            });
        }


        /// <summary>
        /// 同步发送消息
        /// </summary>
        /// <param name="topicChannel">主题通道</param>
        /// <param name="exchangeName">交换名称</param>
        /// <param name="content">文本消息内容</param>
        /// <param name="ququeTimeSpan">队列过期时间(毫秒) </param>
        /// <param name="messageTimeSpan">消息过期时间(毫秒)</param>
        /// <param name="ip">消息队列服务IP</param>
        /// <param name="port">消息队列服务端口</param>
        public static bool Send(string topicChannel, string exchangeName, string content, long? ququeTimeSpan = null, long? messageTimeSpan = null, string ip = null, int port = 5672, string userName = null, string passWord = null, string VirtualHost = null)
        {
            // 连接字符串
            string connectionString = null;
            string username = null;
            string password = null;
            string virtualhost = null;

            // 检查配置
            if (string.IsNullOrEmpty(ip))
            {
                if (string.IsNullOrEmpty(ConnectionString))
                    throw new Exception("appsettings.json does not define HostName url section:[RabbitMQ:HostName]");

                connectionString = ConnectionString;
            }
            else
            {
                connectionString = $"amqp://{ip}:{port}/";
            }

            if (string.IsNullOrEmpty(userName))
            {
                if (string.IsNullOrEmpty(USER_NAME))
                    throw new Exception("appsettings.json does not define UserName url section:[RabbitMQ:UserName]");

                username = USER_NAME;
            }
            else
            {
                username = userName;
            }

            if (string.IsNullOrEmpty(passWord))
            {
                if (string.IsNullOrEmpty(PASS_WORD))
                    throw new Exception("appsettings.json does not define PassWord url section:[RabbitMQ:PassWord]");

                password = PASS_WORD;
            }
            else
            {
                password = userName;
            }

            if (string.IsNullOrEmpty(VirtualHost))
            {
                virtualhost = VIRTUAL_HOST;
            }

            if (string.IsNullOrEmpty(VirtualHost))
            {
                if (string.IsNullOrEmpty(VIRTUAL_HOST))
                    throw new Exception("appsettings.json does not define VirtualHost url section:[RabbitMQ:VirtualHost]");

                virtualhost = VIRTUAL_HOST;
            }
            else
            {
                virtualhost = VirtualHost;
            }


            // 执行发送
            try
            {
                var url = new Uri(connectionString);
                // 创建连接工厂
                IConnectionFactory factory = new ConnectionFactory()
                {
                    Endpoint = new AmqpTcpEndpoint(url),
                    //HostName = connectionString,
                    //Port = port,
                    UserName = username,
                    Password = password,
                    VirtualHost = virtualhost,
                    //RequestedHeartbeat = 0 时间间隔配置为0，则表示不启用heartbeat检测。暂不处理
                };

                using (IConnection connection = factory.CreateConnection())
                {
                    using (var channel = connection.CreateModel())
                    {
                        Dictionary<string, object> queueTime = null;
                        if (ququeTimeSpan != null && ququeTimeSpan > 0)
                        {
                            queueTime = new Dictionary<string, object>();
                            queueTime.Add("x-message-ttl", ququeTimeSpan);
                        }
                        /**
                         * 声明一个队列。
                         * 参数一：队列名称
                         * 参数二：是否持久化
                         * 参数三：是否排外  如果排外则这个队列只允许有一个消费者
                         * 参数四：是否自动删除队列，如果为true表示没有消息也没有消费者连接自动删除队列
                         * 参数五：队列的附加属性
                         * 注意：
                         * 1.声明队列时，如果已经存在则放弃声明，如果不存在则会声明一个新队列；
                         * 2.队列名可以任意取值，但需要与消息接收者一致。
                         * 3.下面的代码可有可无，一定在发送消息前确认队列名称已经存在RabbitMQ中，否则消息会发送失败。
                         */
                        channel.QueueDeclare(topicChannel, true, false, false, queueTime);

                        if (!string.IsNullOrEmpty(exchangeName))
                        {
                            /**
                              * 声明一个交换机名称。
                              * 参数一：交换机名称
                              * 参数二：交换机类型，常见的如fanout、direct、topic
                              * fanout : 广播 -----所有捆绑了得客户都能收到消息。 direct : 组播 -------只有同一组的客户才能收到。 topic : 规则波 ------ 满足某一要求的所有可客户都能收到。
                              * 参数三：设置是否持久化。durable设置true表示持久化，反之是持久化。持久化可以将将换机存盘，在服务器重启时不会丢失相关信息
                              * 参数四：设置是否自动删除。autoDelete设置为true则表示自动删除。自动删除的前提是至少有一个队列或者交换机与这个交换器绑定的队列或者交换器都与之解绑
                              * 注意：
                              * 1.声明队列时，如果已经存在则放弃声明，如果不存在则会声明一个新队列；
                              * 2.队列名可以任意取值，但需要与消息接收者一致。
                              * 3.下面的代码可有可无，一定在发送消息前确认队列名称已经存在RabbitMQ中，否则消息会发送失败。
                              */
                            channel.ExchangeDeclare(exchangeName, "topic", true, false, null);

                            //绑定队列
                            channel.QueueBind(topicChannel, exchangeName, topicChannel);
                        }


                        var properties = channel.CreateBasicProperties();

                        if (messageTimeSpan != null && messageTimeSpan > 0)
                        {
                            //设置消息持久化
                            properties.Persistent = true;

                            //设置当个消息的过期时间为5000毫秒
                            properties.Expiration = Convert.ToString(messageTimeSpan);
                        }

                        //处理需要发送消息
                        var body = Encoding.UTF8.GetBytes(content);

                        /**
                          * 发送消息到MQ
                          * 参数一：交换机名称，为""表示不用交换机
                          * 参数二:为队列名称或者routingKey.当指定了交换机就是routingKey
                          * 参数三：消息的属性信息
                          * 参数四：消息内容的字节数组
                          */
                        channel.BasicPublish(exchangeName, topicChannel, properties, body);

                        return true;
                    }
                }
            }
            catch (Exception ex)
            {
#pragma warning disable CS4014 // 由于此调用不会等待，因此在调用完成前将继续执行当前方法
                LogHelper.Error("Send To RabbitMQ Failed.", ex);
#pragma warning restore CS4014 // 由于此调用不会等待，因此在调用完成前将继续执行当前方法
                return false;
            }
        }


        /// <summary>
        /// 接收消息
        /// </summary>
        /// <param name="topicChannel">主题通道</param>
        /// <param name="clientId">客户端Id，用于持久订阅用，如果不需要持久订阅可以传空</param>
        /// <param name="selector">消息过滤器，与发布消息的filterKey相关，如果接收所有消息则可以传空</param>
        /// <param name="action">接收消息后执行的自定义动作，其中string类型的参数为接收到的文本消息内容</param>
        /// <param name="connectionString">消息队列地址，默认从配置文件获取，如果传入则以传入的为准</param>
        /// <param name="userName">消息队列用户名称，默认从配置文件获取，如果传入则以传入的为准</param>
        /// <param name="passWord">消息队列用户密码，默认从配置文件获取，如果传入则以传入的为准</param>
        /// <param name="virtualHost">消息队列虚拟机，如果传入则以传入的为准</param>
        public static void Receive(string topicChannel, string clientId, string selector, Action<string> action
        , string connectionString = null, string userName = null, string passWord = null, string virtualHost = null)
        {
            // 如果未传入连接地址，则从配置文件获取
            if (string.IsNullOrEmpty(connectionString))
            {
                // 检查配置
                if (string.IsNullOrEmpty(ConnectionString))
                    throw new Exception("appsettings.json does not define HostName url section:[RabbitMQ:HostName]");

                connectionString = ConnectionString;
            }
            // 如果未传入连接地址，则从配置文件获取
            if (string.IsNullOrEmpty(userName))
            {
                // 检查配置
                if (string.IsNullOrEmpty(USER_NAME))
                    throw new Exception("appsettings.json does not define UserName url section:[RabbitMQ:UserName]");

                userName = USER_NAME;
            }
            // 如果未传入连接地址，则从配置文件获取
            if (string.IsNullOrEmpty(passWord))
            {
                // 检查配置
                if (string.IsNullOrEmpty(PASS_WORD))
                    throw new Exception("appsettings.json does not define PassWord url section:[RabbitMQ:PassWord]");

                passWord = PASS_WORD;
            }
            if (string.IsNullOrEmpty(virtualHost))
            {
                virtualHost = VIRTUAL_HOST;
            }


            // 创建连接工厂
            IConnectionFactory factory = new ConnectionFactory()
            {
                Endpoint = new AmqpTcpEndpoint(connectionString),
                UserName = userName,
                Password = passWord,
                VirtualHost = virtualHost,
                //RequestedHeartbeat = 0 时间间隔配置为0，则表示不启用heartbeat检测。暂不处理
            };

            // 创建连接
            IConnection connection = factory.CreateConnection();
            var channel = connection.CreateModel();

            /**
              * 声明一个队列。
              * 参数一：队列名称
              * 参数二：是否持久化
              * 参数三：是否排外  如果排外则这个队列只允许有一个消费者
              * 参数四：是否自动删除队列，如果为true表示没有消息也没有消费者连接自动删除队列
              * 参数五：队列的附加属性
              * 注意：
              * 1.声明队列时，如果已经存在则放弃声明，如果不存在则会声明一个新队列；
              * 2.队列名可以任意取值，但需要与消息接收者一致。
              * 3.下面的代码可有可无，一定在发送消息前确认队列名称已经存在RabbitMQ中，否则消息会发送失败。
              */
            channel.QueueDeclare(topicChannel, true, false, false, null);

            /**
             * 接收消息。会持续坚挺，不能关闭channel和Connection
             * 参数一：队列名称
             * 参数二：消息是否自动确认，true表示自动确认接收完消息以后会自动将消息从队列移除。否则需要手动ack消息
             * 参数三：消息接收者的标签，用于多个消费者同时监听一个队列时用于确认不同消费者。
             * 参数四：消息接收者
             */
            var consumer = new EventingBasicConsumer(channel);//消费者
            channel.BasicConsume(topicChannel, true, consumer);//消费消息
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body.ToArray();
                var message = Encoding.UTF8.GetString(body);

                //执行传入的自定义操作
                action(message);
            };

        }

    }
}
